package org.teremail.smtp;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Collection;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.handler.support.IoSessionInputStream;
import org.teremail.common.Log;
import org.teremail.delivery.Envelope;
import org.teremail.delivery.Exchange;
import org.teremail.message.Address;
import org.teremail.message.Content;
import org.teremail.message.Message;
import org.teremail.message.builder.MessageHeadersBuilder;
import org.teremail.message.builder.SimpleMessageBuilder;
import org.teremail.smtp.commands.SMTPCommand;
import org.teremail.store.Store;
import org.teremail.util.Copier;
import org.teremail.util.DotUnstuffingCopier;

public final class MessageDataConsumer implements DataConsumer {

    public static final String MSG_451 = "451 Requested action aborted: error in processing";
    private final static Log log = Log.getLog(MessageDataConsumer.class);
    private final Exchange exchange;
    private final Store store;
    private final String id;
    private final Thread t;
    private final IoSessionInputStream in;
    private final InputStream bufIn;
    private final Copier copier = new DotUnstuffingCopier();
    private final SMTPSession session;
    private Message m = null;

    public MessageDataConsumer(SMTPSession session, Exchange exchange) {
        this.session = session;
        this.exchange = exchange;
        this.store = exchange.getStore();
        this.id = store.createStoreItem();
        this.in = new IoSessionInputStream();
        this.bufIn = new BufferedInputStream(in);
        t = new Thread(new ParseRunnable());
        t.start();
    }
    
    private class ParseRunnable implements Runnable {

        public void run() {
            try {
                
                Address sender = session.getSender();
                if (sender == null) {
                    throw new IllegalArgumentException(SMTPCommand.MSG_503);
                }
                
                Collection<Address> recipients = session.getRecipients();
                if (recipients.size() == 0) {
                    throw new IllegalArgumentException(SMTPCommand.MSG_554);
                }
                
                SimpleMessageBuilder smb = new SimpleMessageBuilder();
                MessageHeadersBuilder mhb = smb.getHeaderBuilder();
                mhb.load(bufIn);
                OutputStream out = store.getOutputStream(id);
                copier.copy(bufIn, out, 8192);
                out.flush();
                out.close();
                smb.setContent(new Content(id));
                m = smb.build();
                
                Envelope env = new Envelope(sender, recipients, m);
                
                exchange.deliver(env);
                log.info("Message Received");
                session.write(SMTPCommand.MSG_OK);
            } catch (IOException e) {
                // Fail with a temporary error.
                log.error(e, "Failed");
                session.write(MSG_451);
                
            } catch (IllegalArgumentException e) {
                // Fail with a permanent error.
                log.error("Failed create message: %s", e.getMessage());
                session.write(e.getMessage());
                
            } finally {
                session.setDataMode(false);
            }
        }
    }
    
    public void close() throws IOException {
        in.close();
    }

    public void write(ByteBuffer b) {
        in.write(b);
    }

    public Message getMessage() {
        try {
            t.join();
            return m;
        } catch (InterruptedException e) {
            return null;
        }
    }

    Store getStore() {
        return store;
    }

}
